package hanlen.mq.starter.core.impl;

import hanlen.mq.starter.configuration.properties.HanlenMqProperties;
import hanlen.mq.starter.constants.QueueConstants;
import hanlen.mq.starter.core.ThreadPoolExecutorService;
import hanlen.mq.starter.support.QueueUtil;
import hanlen.mq.starter.support.thread.AbortPolicyWithReport;
import hanlen.mq.starter.support.thread.NamedThreadFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * mq-spring-boot-starter
 *
 * @author MrYe
 * @Description
 * @date 2021-10-25 18:54
 */
public class DefaultThreadPoolExecutorImpl implements ThreadPoolExecutorService {

    @Autowired
    private HanlenMqProperties mqProperties;

    private ThreadPoolExecutor threadPoolExecutor = null;

    @Override
    public Executor executor() {
        return threadPoolExecutor();
    }

    @Override
    public ThreadPoolExecutor threadPoolExecutor() {
        if (null == threadPoolExecutor) {
            synchronized (DefaultThreadPoolExecutorImpl.class) {
                if (null == threadPoolExecutor) {
                    Map<String, Integer> threadConfig = MapUtils.isEmpty(mqProperties.getThreads()) ? new HashMap<>(0) : mqProperties.getThreads();
                    final int corePoolSize = QueueUtil.defaultIfNull(threadConfig.get(QueueConstants.CORE_POOL_SIZE), QueueUtil.getProcessors() * 2);
                    final int maxPoolSize = QueueUtil.defaultIfNull(threadConfig.get(QueueConstants.MAX_POOL_SIZE), corePoolSize + 10);
                    final int keepAliveSecond = QueueUtil.defaultIfNull(threadConfig.get(QueueConstants.KEEP_ALIVE_SECOND), 60);
                    final int blockQueueSize = QueueUtil.defaultIfNull(threadConfig.get(QueueConstants.BLOCK_QUEUE_SIZE), 1024);

                    final String prefix = "mq-";
                    threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveSecond, TimeUnit.SECONDS,
                        blockQueueSize == 0 ? new SynchronousQueue<>() : (blockQueueSize < 0 ? new LinkedBlockingQueue<>() : new LinkedBlockingQueue<>(blockQueueSize)),
                        new NamedThreadFactory(prefix, false), new AbortPolicyWithReport(prefix)) {
                    };
                }
            }
        }
        return threadPoolExecutor;
    }

}
